package com.ng.biz.service.impl;

import java.sql.Timestamp;
import java.util.Date;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
import com.jfinal.kit.JsonKit;
import com.jfinal.kit.StrKit;
import com.jfinal.plugin.activerecord.Db;
import com.jfinal.plugin.activerecord.Record;
import com.ng.biz.model.domain.DataLogOutput;
import com.ng.biz.model.domain.Datasource;
import com.ng.biz.service.IDataFlow;
import com.ng.biz.service.IDataLog;
import com.ng.biz.service.IDatasource;
import com.ng.biz.service.IMqttClient;
import com.ng.common.enums.DataLogStatusEnum;
import com.ng.common.util.DateUtil;
import com.ng.common.util.SpringContextHolder;
import com.ng.common.util.UuidUtil;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Service
public class DataFlowImpl implements IDataFlow {
	
	@Autowired
	private StringRedisTemplate redisTemplate;
	
	@Autowired
	private IDatasource datasourceImpl;
	
	
    @Autowired
    private IDataLog dataLogService;
	
	@Override
	public void execute(String dataLogId,String dataLogInputId,JSONObject flow, String json) throws Exception {
		
		if(flow==null || flow.isEmpty()) {
			throw new Exception( "执行流程信息不存在（flow）.");
		}
		
		JSONObject content = flow.getJSONObject("content");
		if(content==null || content.isEmpty()) {
			throw new Exception("执行流程信息不存在（content）.");
		}
		
		String valueFieldIdentifier = content.getString("valueFieldIdentifier");
		String valueFieldType = content.getString("valueFieldType");
		
		String valueFieldJsonReturnType="";
		if(StrKit.equals("map", valueFieldType)) {
			valueFieldJsonReturnType = content.getString("valueFieldJsonReturnType");
		}
		
		
		
		JsonValue jsonValue = getJsonValue(json, valueFieldIdentifier,valueFieldType,valueFieldJsonReturnType);
		if(jsonValue == null) {
			log.error("未找到指定字段: {}", valueFieldIdentifier);
			throw new Exception(String.format("请求结果信息中不包含[%s]字段.", valueFieldIdentifier));
		}
		
		JSONArray standardFields = flow.getJSONArray("standardFields");
		if(standardFields==null || standardFields.isEmpty()) {
			log.error("未找到标准化字段: {}", valueFieldIdentifier);
			throw new Exception(String.format("未找到标准化字段: %s.", valueFieldIdentifier));
		}
		
		// 根据字段类型处理数据
		try {
			List<JSONObject> standards=new LinkedList<JSONObject>();
			switch(jsonValue.getType().name().toLowerCase()) {
				case "struct":
					if(jsonValue.getType() == JsonValueType.OBJECT) {
						// 处理单个对象
						JSONObject data = (JSONObject) jsonValue.getValue();
						standards.add(convertInput(data,standardFields));
					} else {
						//log.error("字段类型不匹配，期望OBJECT类型，实际是: {}", jsonValue.getType());
						throw new Exception(String.format("字段类型不匹配，期望OBJECT类型，实际是: %s", jsonValue.getType()));
					}
					break;
					
				case "object":
					if(jsonValue.getType() == JsonValueType.OBJECT) {
						// 处理单个对象
						JSONObject data = (JSONObject) jsonValue.getValue();
						standards.add(convertInput(data,standardFields));
					} else {
						//log.error("字段类型不匹配，期望OBJECT类型，实际是: {}", jsonValue.getType());
						throw new Exception(String.format("字段类型不匹配，期望OBJECT类型，实际是: %s", jsonValue.getType()));
					}
					break;
					
				case "array":
					if(jsonValue.getType() == JsonValueType.ARRAY) {
						// 处理数组
						JSONArray array = (JSONArray) jsonValue.getValue();
						for(int i = 0; i < array.size(); i++) {
							Object item = array.get(i);
							if(item instanceof JSONObject) {
								standards.add(convertInput( (JSONObject) item,standardFields));
							} else {
								//log.warn("数组元素不是对象类型: {}", item);
								throw new Exception(String.format("数组元素不是对象类型: %s", JsonKit.toJson(item)));
							}
						}
					} else {
						//log.error("字段类型不匹配，期望ARRAY类型，实际是: {}", jsonValue.getType());
						throw new Exception(String.format("字段类型不匹配，期望ARRAY类型，实际是: %s", jsonValue.getType()));
					}
					break;
					
				default:
					//log.error("不支持的字段类型: {}", valueFieldIdentifier);
					throw new Exception(String.format("不支持的字段类型: %s",valueFieldIdentifier));
			}
			
			if(standards.isEmpty()) {
				throw new Exception("数据标准失败，转换结果Empty.");
			}
			processOutputData(dataLogId,dataLogInputId,flow.getJSONArray("output"),standards);
		} catch(Exception e) {
			log.error("处理HTTPOrMQTT数据失败: {}", e.getMessage());
			throw e;
		}
	}
	
	@Override
	public void processOutputData(String dataLogId,String dataLogInputId,JSONArray jsonArray,List<JSONObject> records) throws Exception {
		
		if(jsonArray==null || jsonArray.isEmpty()) {
			throw new Exception("未设置输出源.");
		}
		
		for(Object obj : jsonArray) {
			JSONObject outputObject=(JSONObject)obj;
			
			if(outputObject==null || outputObject.isEmpty()) {
				continue;
			}
			
			String type=outputObject.getString("type");
			String fieldStandardSame=outputObject.getString("fieldStandardSame");
			
			List<JSONObject> outputs=new LinkedList<JSONObject>();
			JSONArray outputFields=null;
			if(!StrKit.equals("1", fieldStandardSame)) {
				outputFields = outputObject.getJSONArray("fields");
				for(JSONObject jsonObject: records) {
					outputs.add(convertOutput(jsonObject,outputFields));
				}
			}else {
				outputs.addAll(records);
			}
			
			if(StrKit.equals("OLTP", type)) {
				String datasourceId=outputObject.getString("datasourceId");
				String table=outputObject.getString("table");
				String primarykey=outputObject.getString("primarykey");
				handleOLTPOutput( dataLogId, dataLogInputId,outputs, datasourceId, table,primarykey);
			}
			else if(StrKit.equals("HTTP", type)) {
				//String datasourceId=outputObject.getString("datasourceId");
				//handleHTTPOutput(outputs, datasourceId);
			}
			else if(StrKit.equals("MQTT", type)) {
				handleMQTTOutput(outputObject,outputs);
			}
			else if(StrKit.equals("REDIS", type)) {
				handleRedisOutput(outputObject,outputs);
			}
			else if(StrKit.equals("SERIES", type)) {
				handleSeriesOutput(outputObject,outputFields,outputs);
			}
		}
	}
	
	private void handleSeriesOutput(JSONObject outputObject,JSONArray outputFields, List<JSONObject> outputs) throws Exception {
		String datasourceId=outputObject.getString("datasourceId");
		String table=outputObject.getString("table");
		String primarykey=outputObject.getString("primarykey");
		Datasource datasource=Datasource.dao.findById(datasourceId);
		if(datasource!=null) {
			if(StrKit.equals("InfluxDB", datasource.getType())) {
				InfluxDBClient influxDBClient=datasourceImpl.getInfluxDBClient(datasourceId);
				if(influxDBClient==null) {
					return;
				}
				for(JSONObject data : outputs) {
					try {
						Point point = Point.measurement(String.format("%s%s",table,data.get(primarykey)));
						for(Object obj:outputFields) {
							JSONObject field=(JSONObject)obj;
							if(StrKit.equals("field", field.getString("outputValueStorageType"))) {
								String outputFieldIdentifier=field.getString("outputFieldIdentifier");
								switch(field.getString("outputFieldType")) {
									case "int" -> point.addField(outputFieldIdentifier, data.getInteger(outputFieldIdentifier)) ;
									case "long" -> point.addField(outputFieldIdentifier, Long.parseLong(data.getString(outputFieldIdentifier)));
									case "float" -> point.addField(outputFieldIdentifier, Float.parseFloat(data.getString(outputFieldIdentifier)));
									case "double" -> point.addField(outputFieldIdentifier, Double.parseDouble(data.getString(outputFieldIdentifier)));
									case "boolean" -> point.addField(outputFieldIdentifier, Boolean.parseBoolean(data.getString(outputFieldIdentifier)));
									case "dateString" -> point.addField(outputFieldIdentifier, data.getString(outputFieldIdentifier)); // 日期字符串已在之前处理
									case "timestampString" -> point.addField(outputFieldIdentifier, data.getString(outputFieldIdentifier)); // 时间戳已在之前处理
									default -> point.addField(outputFieldIdentifier, data.getString(outputFieldIdentifier));
								};
							}
							else if(StrKit.equals("tag", field.getString("outputValueStorageType"))) {
								point.addTag(field.getString("outputFieldIdentifier"), data.getString(field.getString("outputFieldIdentifier")));
							}
							else if(StrKit.equals("time", field.getString("outputValueStorageType"))) {
								point.time(new Timestamp(Long.valueOf(data.getString(field.getString("outputFieldIdentifier")))).toInstant(), WritePrecision.S); // 使用毫秒级时间戳
							}
						}
						influxDBClient.getWriteApiBlocking().writePoint(point);
					} catch (Exception e) {
						throw new Exception(String.format("InfluxDB输出异常:%s", e.getMessage()));
					}
				}
			}
		}
	}
	
	private void handleOLTPOutput(String dataLogId,String dataLogInputId,List<JSONObject> outputs, String datasourceId, String table, String primarykey) throws Exception {
		
		DataLogOutput dataLogOutput=null;
		if(!StrKit.isBlank(dataLogId) && !StrKit.isBlank(dataLogInputId) ) {
			 dataLogOutput=dataLogService.createDataLogOutput(dataLogId,dataLogInputId,"OLTP", outputs.size());
		}
		
		if(StrKit.isBlank(primarykey)) {
			if(!StrKit.isBlank(dataLogId) && !StrKit.isBlank(dataLogInputId) && dataLogOutput!=null) {
				dataLogService.buildUpdateDataLogOutput(dataLogOutput.getId())
				.setStatus(DataLogStatusEnum.Error.getType())
				.setInfo("OLTP输出未设置primarykey.")
				.setEndTime(new Date(System.currentTimeMillis()))
				.update();
			}
			throw new Exception("OLTP输出未设置primarykey.");
		}
		
		// 解析主键字段
		String[] primaryKeys = parsePrimaryKeys(primarykey);
		
		if(primaryKeys==null || primaryKeys.length==0) {
			if(!StrKit.isBlank(dataLogId) && !StrKit.isBlank(dataLogInputId)  && dataLogOutput!=null) {
				dataLogService.buildUpdateDataLogOutput(dataLogOutput.getId())
				.setStatus(DataLogStatusEnum.Error.getType())
				.setInfo("OLTP输出未设置primarykey.")
				.setEndTime(new Date(System.currentTimeMillis()))
				.update();
			}
			throw new Exception("OLTP输出未设置primarykey.");
		}
		
		try {
			
			List<Record> inserts=new LinkedList<Record>();
			List<Record> updates=new LinkedList<Record>();
			
			// 批量处理数据
			for(JSONObject data : outputs) {
				// 构建主键条件
				Object[] pkValues = new Object[primaryKeys.length];
				for(int i = 0; i < primaryKeys.length; i++) {
					pkValues[i] = data.get(primaryKeys[i]);
					if(pkValues[i] == null) {
						log.error("主键值不能为空: {}", primaryKeys[i]);
						continue;
					}
				}
				
				// 查询是否存在记录
				Record existingRecord=Db.use(datasourceId).findByIds(table, primarykey, pkValues);
				if(existingRecord != null) {
					// 更新记录
					updates.add(buildUpdateRecord(data));
				} else {
					// 插入记录
					inserts.add(buildInsertRecord(data));
				}
			}
			
			DataLogOutput dataLogOutputUpdate=null;
			if(!StrKit.isBlank(dataLogId) && !StrKit.isBlank(dataLogInputId)  && dataLogOutput!=null) {
				dataLogOutputUpdate=dataLogService.buildUpdateDataLogOutput(dataLogOutput.getId());
			}
			if(inserts.size()>0) {
				int[] batchSave=Db.use(datasourceId).batchSave(table, inserts, inserts.size());
				if(dataLogOutputUpdate!=null) dataLogOutputUpdate.setInsertTotal(batchSave.length);
			}
			if(updates.size()>0) {
				int[] batchUpdate=Db.use(datasourceId).batchUpdate(table, primarykey, updates, updates.size());
				if(dataLogOutputUpdate!=null) dataLogOutputUpdate.setUpdateTotal(batchUpdate.length);
			}
			
			if(dataLogOutputUpdate!=null) {
				dataLogOutputUpdate.setStatus(DataLogStatusEnum.Success.getType());
				dataLogOutputUpdate.setInfo("OLTP输出完成.");
				dataLogOutputUpdate.setEndTime(new Date(System.currentTimeMillis()));
				dataLogOutputUpdate.update();
			}
			
			
		} catch(Exception e) {
			log.error("数据输出异常: {}", e.getMessage());
			
			if(!StrKit.isBlank(dataLogId) && !StrKit.isBlank(dataLogInputId)  && dataLogOutput!=null) {
				dataLogService.buildUpdateDataLogOutput(dataLogOutput.getId())
				.setStatus(DataLogStatusEnum.Error.getType())
				.setInfo(String.format("OLTP输出异常:%s", e.getMessage()))
				.setEndTime(new Date(System.currentTimeMillis()))
				.update();
			}
			throw new Exception(String.format("OLTP输出异常:%s", e.getMessage()));
		}
	}
	
	private void handleRedisOutput(JSONObject outputObject,List<JSONObject> outputs) throws Exception {
		String fixedRedisKey=outputObject.getString("fixedRedisKey");
		
		String redisValueType=outputObject.getString("redisValueType");
		String redisValueField=outputObject.getString("redisValueField");
		
		
		
		for(JSONObject object:outputs) {
			if(StrKit.equals("0", fixedRedisKey)) {
				String redisKeyValueField=outputObject.getString("redisKeyValueField");
				if(object.containsKey(redisKeyValueField)) {
					
					if(StrKit.equals("0", redisValueType)) {
						redisTemplate.opsForValue().set(object.getString(redisKeyValueField), JSONObject.toJSONString(object));
					}
					else {
						if(object.containsKey(redisValueField)) {
							redisTemplate.opsForValue().set(object.getString(redisKeyValueField), object.getString(redisValueField));
						}else {
							throw new Exception("输出Redis是取值异常,输出对象不含"+redisValueField+"字段");
						}
					}
				}
			}else {
				if(StrKit.equals("0", redisValueType)) {
					redisTemplate.opsForValue().set(outputObject.getString("fixedRedisKeyValue"), JSONObject.toJSONString(object));
				}
				else {
					if(object.containsKey(redisValueField)) {
						redisTemplate.opsForValue().set(outputObject.getString("fixedRedisKeyValue"),object.getString(redisValueField));
					}else {
						throw new Exception("输出Redis是取值异常,输出对象不含"+redisValueField+"字段");
					}
				}
			}
		}
	}
	
	private void handleMQTTOutput(JSONObject outputObject,List<JSONObject> outputs) {
		String datasourceId = outputObject.getString("datasourceId");
		String topicTemplate = outputObject.getString("topic");
		int qos = outputObject.getIntValue("qos");
		boolean retained = StrKit.equals("1",outputObject.getString("retained"));
		IMqttClient mqttClientService = SpringContextHolder.getBean(MqttClientImpl.class);
		
		if(mqttClientService != null) {
			for(JSONObject object : outputs) {
				// 处理动态topic
				String topic = replaceDynamicTopic(topicTemplate, object);
				// 发布消息
				mqttClientService.publish(datasourceId, topic, JSONObject.toJSONString(object), qos, retained);
			}
		}
	}
	
	/**
	 * 处理动态topic，替换${field}为实际值
	 */
	private String replaceDynamicTopic(String topicTemplate, JSONObject data) {
		if(StrKit.isBlank(topicTemplate) || data == null) {
			return topicTemplate;
		}
		
		// 匹配所有${field}模式的占位符
		String result = topicTemplate;
		int startIndex = 0;
		while(true) {
			int start = result.indexOf("${", startIndex);
			if(start == -1) break;
			
			int end = result.indexOf("}", start);
			if(end == -1) break;
			
			// 提取字段名
			String fieldName = result.substring(start + 2, end);
			// 获取字段值
			Object fieldValue = data.get(fieldName);
			String replacement = fieldValue != null ? fieldValue.toString() : "";
			
			// 替换占位符
			result = result.substring(0, start) + replacement + result.substring(end + 1);
			startIndex = start + replacement.length();
		}
		
		return result;
	}
	
	/**
	 * 转换输出数据
	 * @param data 输入数据
	 * @param outputFields 输出字段配置
	 * @return 转换后的输出数据
	 */
	private JSONObject convertOutput(JSONObject data, JSONArray outputFields) {
		if(data == null || outputFields == null) {
			return null;
		}
		
		JSONObject result = new JSONObject();
		try {
			for(int i = 0; i < outputFields.size(); i++) {
				JSONObject field = outputFields.getJSONObject(i);
				
				// 获取输入字段信息
				String inputFieldIdentifier = field.getString("inputFieldIdentifier");
				String inputFieldType = field.getString("inputFieldType");
				String inputFieldFormat = field.getString("inputFieldFormat");
				
				// 获取输出字段信息
				String outputFieldIdentifier = field.getString("outputFieldIdentifier");
				String outputFieldType = field.getString("outputFieldType");
				String outputFieldFormat = field.getString("outputFieldFormat");
				String outputFieldDefaultValue = field.getString("outputFieldDefaultValue");
				String outputValueFromInput = field.getString("outputValueFromInput");
				String outputFieldDecimalPoint = field.getString("outputFieldDecimalPoint");
				
				// 处理字段标识符相同性
				if("1".equals(field.getString("sameFieldIdentifier"))) {
					outputFieldIdentifier = inputFieldIdentifier;
				}
				
				// 获取值
				Object value = null;
				
				switch(outputValueFromInput) {
					case "input":
						// 从输入数据获取值
						value = data.get(inputFieldIdentifier);
					if(value != null) {
							// 处理日期格式
							if("dateString".equals(inputFieldType)) {
								String inputDateFormat = "select".equals(field.getString("inputFieldDateFormatSrc")) ? 
									field.getString("inputFieldDateFormat") : inputFieldFormat;
								Date date = DateUtil.dateStrToDate(value.toString(), inputDateFormat);
								if(date != null) {
									value = date;
								}
							}
							// 处理时间戳
							else if("timestampString".equals(inputFieldType)) {
								try {
									long timestamp = Long.parseLong(value.toString());
									value = new Date(timestamp);
								} catch(NumberFormatException e) {
									log.warn("时间戳转换失败: {}", value);
								}
							}
						}
						break;
						
					case "constant":
						// 使用常量值
						value = outputFieldDefaultValue;
						break;
						
					case "uuid":
						value = UuidUtil.getUUID32();
						break;
						
					case "curSystemDate":
						// 获取当前系统时间
						value = new Date();
						break;
						
					case "OLTP":
						// 从数据库获取值
						String datasourceId = field.getString("outputValueDatasourceId");
						String sql = field.getString("outputValueSql");
						if(StrKit.notBlank(sql)) {
							// 替换SQL中的参数
							Object inputValue = data.get(inputFieldIdentifier);
							if(inputValue != null) {
								if(inputValue instanceof Number) {
									sql = sql.replace("#{value}", inputValue.toString());
								} else {
									sql = sql.replace("#{value}", "'" + inputValue.toString() + "'");
								}
							}
							value = Db.use(datasourceId).queryStr(sql);
						}
						break;
						
					case "redis":
						// 从Redis获取值
						String redisKey = field.getString("outputValueRedisKey");
						if(StrKit.notBlank(redisKey)) {
							value = redisTemplate.opsForValue().get(redisKey);
						}
						break;
				}
				
				// 如果值为空且有默认值，使用默认值
				if(value == null && StrKit.notBlank(outputFieldDefaultValue)) {
					value = outputFieldDefaultValue;
				}
				
				// 设置输出值
				if(value != null && StrKit.notBlank(outputFieldIdentifier)) {
					// 进行最终的类型转换
					value = convertValueType(value, "string", null, 
							outputFieldType, outputFieldFormat, outputFieldDecimalPoint);
					result.put(outputFieldIdentifier, value);
				}
		}
		
		return result;
		} catch(Exception e) {
			log.error("Convert output data failed", e);
			return null;
		}
	}
	
	/**
	 * 构建更新记录
	 */
	private Record buildUpdateRecord(JSONObject data) {
		Record record = new Record();
		for(String key : data.keySet()) {
			record.set(key, data.get(key));
		}
		return record;
	}
	
	/**
	 * 构建插入记录
	 */
	private Record buildInsertRecord(JSONObject data) {
		return buildUpdateRecord(data);
	}
	
	/**
	 * 解析主键字段
	 * @param primarykey 主键配置字符串，多个字段用英文.分隔
	 * @return 主键字段数组
	 */
	private String[] parsePrimaryKeys(String primarykey) {
		if(StrKit.isBlank(primarykey)) {
			return new String[0];
		}
		
		// 分割并去除空白字符
		String[] keys = primarykey.split("\\.");
		for(int i = 0; i < keys.length; i++) {
			keys[i] = keys[i].trim();
		}
		
		// 过滤空字段
		return java.util.Arrays.stream(keys)
			.filter(StrKit::notBlank)
			.toArray(String[]::new);
	}
	
	
	/**
	 * 转换输入数据
	 * @param data 输入数据
	 * @param standardFields 标准字段配置
	 * @return 转换后的标准数据
	 */
	private JSONObject convertInput(JSONObject data, JSONArray standardFields) {
		if(data == null || standardFields == null) {
			return null;
		}
		
		JSONObject result = new JSONObject();
		try {
			for(int i = 0; i < standardFields.size(); i++) {
				JSONObject field = standardFields.getJSONObject(i);
				
				// 1. 输入字段配置
				//String inputFieldName = field.getString("inputFieldName");
				String inputFieldIdentifier = field.getString("inputFieldIdentifier");
				String inputFieldType = field.getString("inputFieldType");
				String inputFieldDateFormatSrc = field.getString("inputFieldDateFormatSrc");
				String inputFieldDateFormat = field.getString("inputFieldDateFormat");
				String inputFieldFormat = field.getString("inputFieldFormat");
				
				// 2. 输出字段配置
				//String outputFieldName = field.getString("outputFieldName");
				String outputFieldIdentifier = field.getString("outputFieldIdentifier");
				String outputFieldType = field.getString("outputFieldType");
				String outputFieldFormat = field.getString("outputFieldFormat");
				String outputFieldDefaultValue = field.getString("outputFieldDefaultValue");
				String outputFieldDecimalPoint = field.getString("outputFieldDecimalPoint");
				String outputValueFromInput = field.getString("outputValueFromInput");
				
				// 3. 处理字段名和标识符相同性
				if("1".equals(field.getString("sameFieldName"))) {
					//outputFieldName = inputFieldName;
				}
				if("1".equals(field.getString("sameFieldIdentifier"))) {
					outputFieldIdentifier = inputFieldIdentifier;
				}
				
				// 4. 获取输入值并处理
				Object inputValue = data.get(inputFieldIdentifier);
				if(inputValue != null) {
					// 4.1 处理日期类型
					if("dateString".equals(inputFieldType)) {
						String dateFormat = "select".equals(inputFieldDateFormatSrc) ? 
							inputFieldDateFormat : inputFieldFormat;
						Date date = DateUtil.dateStrToDate(inputValue.toString(), dateFormat);
						if(date != null) {
							inputValue = date;
						}
					}
				}
				
				// 5. 根据输出值来源处理
				Object finalValue = null;
				switch(outputValueFromInput) {
					case "input" -> finalValue = inputValue;
					case "constant" -> finalValue = outputFieldDefaultValue;
					case "uuid" -> finalValue=UuidUtil.getUUID32();
					case "curSystemDate" -> {
						if(List.of("dateString", "long", "timestampString").contains(outputFieldType)) {
							finalValue = new Date();
						}
					}
					case "OLTP" -> {
						String datasourceId = field.getString("outputValueDatasourceId");
						String sql = field.getString("outputValueSql");
						if(StrKit.notBlank(sql)) {
							// 替换SQL中的参数
							String columnName = inputFieldIdentifier.contains(".") ? 
								inputFieldIdentifier.substring(inputFieldIdentifier.lastIndexOf(".") + 1) : 
								inputFieldIdentifier;
							sql = sql.replace("${" + columnName + "}", 
								inputValue != null ? inputValue.toString() : "");
							finalValue = Db.use(datasourceId).queryStr(sql);
						}
					}
					
					case "redis" -> {
						String redisKey = field.getString("outputValueRedisKey");
						if(StrKit.notBlank(redisKey)) {
							finalValue = redisTemplate.opsForValue().get(redisKey);
						}
					}
				}
				
				// 6. 处理默认值
				if(finalValue == null && StrKit.notBlank(outputFieldDefaultValue)) {
					finalValue = outputFieldDefaultValue;
				}
				
				// 7. 类型转换和输出
				if(finalValue != null && StrKit.notBlank(outputFieldIdentifier)) {
					// 7.1 根据输出字段类型进行转换
					switch(outputFieldType.toLowerCase()) {
						case "string" -> finalValue = finalValue.toString();
						
						case "int" -> finalValue = Integer.parseInt(finalValue.toString());
						
						case "float", "double" -> {
							if(StrKit.notBlank(outputFieldDecimalPoint)) {
								int decimalPoints = Integer.parseInt(outputFieldDecimalPoint);
								double value = Double.parseDouble(finalValue.toString());
								finalValue = String.format("%." + decimalPoints + "f", value);
							}
						}
						
						case "boolean" -> finalValue = Boolean.parseBoolean(finalValue.toString());
						
						case "datestring" -> {
							if(finalValue instanceof Date) {
								finalValue = DateUtil.getFormatDate((Date)finalValue, 
									StrKit.notBlank(outputFieldFormat) ? outputFieldFormat : "yyyy-MM-dd HH:mm:ss");
							}
						}
					}
					
					result.put(outputFieldIdentifier, finalValue);
				}
			}
			
			return result;
		} catch(Exception e) {
			log.error("Convert input data failed", e);
			return null;
		}
	}
	
	/**
	 * JSON值类型枚举
	 */
	private enum JsonValueType {
		STRING, NUMBER, BOOLEAN, ARRAY, OBJECT, NULL
	}
	
	/**
	 * JSON值包装类
	 */
	private static class JsonValue {
		private final Object value;
		private final JsonValueType type;
		
		public JsonValue(Object value, JsonValueType type) {
			this.value = value;
			this.type = type;
		}
		
		public Object getValue() {
			return value;
		}
		
		public JsonValueType getType() {
			return type;
		}
	}
	
	/**
	 * 获取JSON中指定key的值和类型
	 * @param jsonData JSON数据字符串
	 * @param key 要查找的key，支持点号分隔的多级路径，如"data.items.name"
	 *            如果key为空，则直接返回jsonData的类型和值
	 * @return 包含值和类型的JsonValue对象，如果未找到则返回null
	 */
	private JsonValue getJsonValue(String jsonData, String valueFieldIdentifier,String valueFieldType,String valueFieldJsonReturnType) {
		if(StrKit.isBlank(jsonData)) {
			return null;
		}
		
		try {
			// 解析JSON
			Object jsonObj = JSON.parse(jsonData);
			
			// 如果key为空，直接返回jsonData的类型和值
			if(StrKit.isBlank(valueFieldIdentifier)) {
				
				if(StrKit.equals("map", valueFieldType)) {
					if(StrKit.equals("key", valueFieldJsonReturnType)) {
						JSONArray jsonArray=new JSONArray(); 
						JSONObject jsonObject=(JSONObject)jsonObj;
						Iterator<String> it=jsonObject.keySet().iterator();
						String jsonKey = null;
						while(it.hasNext()) {
							jsonKey=it.next();
							jsonArray.add(jsonKey);
						}
						return determineJsonValueType(jsonArray);
					}
					else if(StrKit.equals("value", valueFieldJsonReturnType)) {
						JSONArray jsonArray=new JSONArray(); 
						JSONObject jsonObject=(JSONObject)jsonObj;
						Iterator<String> it=jsonObject.keySet().iterator();
						String jsonKey = null;
						while(it.hasNext()) {
							jsonKey=it.next();
							jsonArray.add(jsonObject.get(jsonKey));
						}
						return determineJsonValueType(jsonArray);
					}
				}
				return determineJsonValueType(jsonObj);
			}
			
			String[] keys = valueFieldIdentifier.split("\\.");
			
			// 逐级查找
			Object current = jsonObj;
			for(String k : keys) {
				if(current == null) {
					return new JsonValue(null, JsonValueType.NULL);
				}
				if(current instanceof JSONArray) {
					// 如果是数组，尝试解析索引
					try {
						int index = Integer.parseInt(k);
						JSONArray array = (JSONArray) current;
						if(index >= 0 && index < array.size()) {
							current = array.get(index);
						} else {
							return null;
						}
					} catch(NumberFormatException e) {
						return null;
					}
				} else if(current instanceof JSONObject) {
					// 如果是对象，直接获取属性
					current = ((JSONObject) current).get(k);
				} else {
					return null;
				}
			}
			
			if(StrKit.equals("map", valueFieldType)) {
				if(StrKit.equals("key", valueFieldJsonReturnType)) {
					JSONArray jsonArray=new JSONArray(); 
					JSONObject jsonObject=(JSONObject)current;
					Iterator<String> it=jsonObject.keySet().iterator();
					String jsonKey = null;
					while(it.hasNext()) {
						jsonKey=it.next();
						jsonArray.add(jsonKey);
					}
					current=jsonArray;
					return determineJsonValueType(jsonArray);
				}
				else if(StrKit.equals("value", valueFieldJsonReturnType)) {
					JSONArray jsonArray=new JSONArray(); 
					JSONObject jsonObject=(JSONObject)current;
					Iterator<String> it=jsonObject.keySet().iterator();
					String jsonKey = null;
					while(it.hasNext()) {
						jsonKey=it.next();
						jsonArray.add(jsonObject.get(jsonKey));
					}
					current=jsonArray;
					return determineJsonValueType(jsonArray);
				}
			}
			return determineJsonValueType(current);
		} catch(Exception e) {
			log.error("解析JSON失败: {}", e.getMessage());
			return null;
		}
	}
	
	
	/**
	 * 确定JSON值的类型
	 * @param value 要判断类型的值
	 * @return JsonValue对象
	 */
	private JsonValue determineJsonValueType(Object value) {
		if(value == null) {
			return new JsonValue(null, JsonValueType.NULL);
		} else if(value instanceof String) {
			return new JsonValue(value, JsonValueType.STRING);
		} else if(value instanceof Number) {
			return new JsonValue(value, JsonValueType.NUMBER);
		} else if(value instanceof Boolean) {
			return new JsonValue(value, JsonValueType.BOOLEAN);
		} else if(value instanceof JSONArray) {
			return new JsonValue(value, JsonValueType.ARRAY);
		} else if(value instanceof JSONObject) {
			return new JsonValue(value, JsonValueType.OBJECT);
		}
		
		return new JsonValue(value.toString(), JsonValueType.STRING);
	}

	public static void main(String[] args) {
		String json = "{\r\n"
				+ "  \"123456\": {\r\n"
				+ "    \"total\": 23,\r\n"
				+ "    \"time\": \"2025-02-09 16:44:29\",\r\n"
				+ "    \"card_id\": \"Test_001\"\r\n"
				+ "  }\r\n"
				+ "}";

		DataFlowImpl task = new DataFlowImpl();
		
//		// 获取简单值
		//JsonValue status = task.getJsonValue(json, "$key:value");
		//System.out.println("total: " + status.getValue() + ", type: " + status.getType());
//		
//		// 获取数组
		JsonValue lives = task.getJsonValue(json, "","map","value");
		System.out.println("lives type: " + lives.value);
//		
//		// 获取嵌套值
//		JsonValue province = task.getJsonValue(json, "lives.0.province");
//		System.out.println("province: " + province.getValue() + ", type: " + province.getType());
//		
//		String topicTemplate="/xyky/xpd/car/real/${carId}";
//		JSONObject jsonObject=new JSONObject();
//		jsonObject.put("carId", "123");
//		String abc=task.replaceDynamicTopic( topicTemplate, jsonObject);
//		System.out.println(abc);
		
	}

	/**
	 * 转换值类型
	 */
	public Object convertValueType(Object value, String inputType, String inputFormat,
			String outputType, String outputFormat, String decimalPoint) {
		if(value == null) return null;
		
		try {
			// 如果输入是日期对象，直接用于后续转换
			if(value instanceof Date) {
				return convertDateToTargetType((Date)value, outputType, outputFormat);
			}

			// 根据输入类型进行初步转换
			Object convertedValue = switch(inputType) {
				case "int" -> Integer.parseInt(value.toString());
				case "long" -> Long.parseLong(value.toString());
				case "float" -> Float.parseFloat(value.toString());
				case "double" -> Double.parseDouble(value.toString());
				case "boolean" -> Boolean.parseBoolean(value.toString());
				case "dateString" -> value; // 日期字符串已在之前处理
				case "timestampString" -> value; // 时间戳已在之前处理
				default -> value.toString();
			};

			// 根据输出类型进行最终转换
			return convertToTargetType(convertedValue, outputType, outputFormat, decimalPoint);
		} catch(Exception e) {
			log.error("类型转换失败: {} -> {}", value, outputType, e);
			return value;
		}
	}
	
	/**
	 * 将日期对象转换为目标类型
	 */
	private Object convertDateToTargetType(Date date, String targetType, String format) {
		if(date == null) return null;
		
		return switch(targetType) {
			case "dateString" -> {
				String dateFormat = StrKit.notBlank(format) ? format : "yyyy-MM-dd HH:mm:ss";
				yield DateUtil.getFormatDate(date, dateFormat);
			}
			case "timestampString" -> String.valueOf(date.getTime());
			default -> date;
		};
	}

	/**
	 * 转换为输出类型
	 */
	private Object convertToTargetType(Object value, String outputType, String outputFormat, String decimalPoint) {
		if(value == null) return null;
		
		try {
			switch(outputType.toLowerCase()) {
				case "string":
					return value.toString();
					
				case "int":
					if(value instanceof Integer) return value;
					return Integer.parseInt(value.toString());
					
				case "long":
					if(value instanceof Long) return value;
					return Long.parseLong(value.toString());
					
				case "float":
					if(value instanceof Float) {
						if(StrKit.notBlank(decimalPoint)) {
							int decimalPoints = Integer.parseInt(decimalPoint);
							return String.format("%." + decimalPoints + "f", (Float)value);
						}
						return value;
					}
					float floatValue = Float.parseFloat(value.toString());
					if(StrKit.notBlank(decimalPoint)) {
						int decimalPoints = Integer.parseInt(decimalPoint);
						return String.format("%." + decimalPoints + "f", floatValue);
					}
					return floatValue;
					
				case "double":
					if(value instanceof Double) {
						if(StrKit.notBlank(decimalPoint)) {
							int decimalPoints = Integer.parseInt(decimalPoint);
							return String.format("%." + decimalPoints + "f", (Double)value);
						}
						return value;
					}
					double doubleValue = Double.parseDouble(value.toString());
					if(StrKit.notBlank(decimalPoint)) {
						int decimalPoints = Integer.parseInt(decimalPoint);
						return String.format("%." + decimalPoints + "f", doubleValue);
					}
					return doubleValue;
					
				case "boolean":
					if(value instanceof Boolean) return value;
					return Boolean.parseBoolean(value.toString());
					
				case "datestring":
					if(value instanceof Date) {
						if(StrKit.notBlank(outputFormat)) {
							return DateUtil.getFormatDate((Date)value, outputFormat);
						}
						return DateUtil.getFormatDate((Date)value, "yyyy-MM-dd HH:mm:ss");
					}
					if(StrKit.notBlank(outputFormat)) {
						Date date = DateUtil.dateStrToDate(value.toString());
						return DateUtil.getFormatDate(date, outputFormat);
					}
					return value.toString();
					
				case "timestampstring":
					if(value instanceof Date) {
						// 返回时间戳字符串
						return String.valueOf(((Date)value).getTime());
					}
					if(value instanceof Number) {
						// 如果已经是数字，直接转字符串
						return value.toString();
					}
					// 尝试解析日期字符串，然后转为时间戳
					try {
						Date date = DateUtil.dateStrToDate(value.toString());
						return String.valueOf(date.getTime());
					} catch(Exception e) {
						log.error("时间戳转换失败: {}", value, e);
						return null;
					}
					
				default:
					return value;
			}
		} catch(Exception e) {
			log.error("输出类型转换失败: {} -> {}", value, outputType, e);
			return null;
		}
	}
}
